--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 50fb3cbfc863151aab5ea3e37bf58cc620650089
Parents : e3a56d5
Author : Ivan <ivan@quad4.io>
Signature : Invalid signer <e46112d44649266d71fe2193e00a4710>, author is <ivan@quad4.io>
Date : 2026-07-09T13:07:42-05:00
feat(interface-sanitization): implement interface name sanitization to prevent ConfigObj errors
Changes
6 files changed, 314 insertions(+), 42 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index 538fc878..34fcff19 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -341,35 +341,6 @@ def _install_reticulum_signal_handlers() -> bool:
return False
-def _python_jit_status_line() -> str:
- jit_runtime = getattr(sys, "_jit", None)
- if jit_runtime is None:
- return "Python JIT: unavailable"
-
- is_available = getattr(jit_runtime, "is_available", None)
- if not callable(is_available):
- return "Python JIT: unavailable"
-
- try:
- available = bool(is_available())
- except Exception:
- return "Python JIT: unavailable"
-
- if not available:
- return "Python JIT: unavailable"
-
- is_enabled = getattr(jit_runtime, "is_enabled", None)
- if not callable(is_enabled):
- return "Python JIT: disabled"
-
- try:
- enabled = bool(is_enabled())
- except Exception:
- enabled = False
-
- return "Python JIT: enabled" if enabled else "Python JIT: disabled"
-
-
def list_host_network_interfaces():
"""Enumerate kernel network interfaces on the host running MeshChat.
@@ -3774,14 +3745,69 @@ class ReticulumMeshChat:
snapshot[name] = {}
return snapshot
- def _write_reticulum_config(self):
+ def _sanitize_interfaces_section_names(self) -> None:
+ """Rewrite interface section keys so ConfigObj can reload the config file."""
+ interfaces = self._get_interfaces_section()
+ if not isinstance(interfaces, dict) or not interfaces:
+ return
+ renamed: dict = {}
+ changed = False
+ for name, details in list(interfaces.items()):
+ safe = InterfaceEditor.sanitize_interface_section_name(name)
+ if not safe:
+ safe = "Interface"
+ if safe != name:
+ changed = True
+ # Avoid collisions after sanitizing two distinct names to the same key.
+ final = safe
+ suffix = 2
+ while final in renamed:
+ final = f"{safe} ({suffix})"
+ suffix += 1
+ if final != name:
+ changed = True
+ renamed[final] = details
+ if changed:
+ if hasattr(self, "reticulum") and self.reticulum:
+ self.reticulum.config["interfaces"] = renamed
+
+ def _verify_reticulum_config_reloadable(self) -> None:
+ """Ensure the on-disk config still parses after write (ConfigObj quirk)."""
+ from RNS.vendor.configobj import ConfigObj
+
+ path = None
+ try:
+ path = self._api_reticulum_config_path()
+ except Exception:
+ path = None
+ if not path:
+ path = getattr(getattr(self, "reticulum", None), "configpath", None)
+ if not path or not os.path.isfile(path):
+ return
+ ConfigObj(path)
+
+ def _write_reticulum_config(self, *, rollback_interfaces=None):
try:
if hasattr(self, "reticulum") and self.reticulum:
+ self._sanitize_interfaces_section_names()
self.reticulum.config.write()
+ self._verify_reticulum_config_reloadable()
return True
return False
except Exception as e:
print(f"Failed to write Reticulum config: {e}")
+ if (
+ rollback_interfaces is not None
+ and hasattr(self, "reticulum")
+ and self.reticulum
+ ):
+ try:
+ self.reticulum.config["interfaces"] = rollback_interfaces
+ except Exception as restore_exc:
+ print(
+ "Failed to restore Reticulum interfaces after write error: "
+ f"{restore_exc}"
+ )
return False
def _detect_failed_autointerfaces(self):
@@ -5854,7 +5880,9 @@ class ReticulumMeshChat:
async def reticulum_interfaces_add(request):
# get request data
data = await request.json()
- interface_name = data.get("name")
+ interface_name = InterfaceEditor.sanitize_interface_section_name(
+ data.get("name")
+ )
interface_type = data.get("type")
allow_overwriting_interface = data.get("allow_overwriting_interface", False)
@@ -6680,12 +6708,19 @@ class ReticulumMeshChat:
InterfaceEditor.update_value(interface_details, data, "ifac_size")
# merge new interface into existing interfaces
+ interfaces_before_write = self._get_interfaces_snapshot()
interfaces[interface_name] = interface_details
# save config
- if not self._write_reticulum_config():
+ if not self._write_reticulum_config(
+ rollback_interfaces=interfaces_before_write
+ ):
return web.json_response(
{
- "message": "Failed to write Reticulum config",
+ "message": (
+ "Failed to write Reticulum config. "
+ "Interface names must not contain '[' or ']' "
+ "(ConfigObj section syntax)."
+ ),
},
status=500,
)
@@ -6809,7 +6844,16 @@ class ReticulumMeshChat:
interface_config = {}
for interface in selected_interfaces:
# add interface and keys/values
- interface_name = interface["name"]
+ interface_name = InterfaceEditor.sanitize_interface_section_name(
+ interface.get("name")
+ )
+ if not interface_name:
+ return web.json_response(
+ {
+ "message": "Imported interface is missing a valid name",
+ },
+ status=422,
+ )
interface_config[interface_name] = {}
for key, value in interface.items():
interface_config[interface_name][key] = value
@@ -6882,12 +6926,19 @@ class ReticulumMeshChat:
)
# update reticulum config with new interfaces
+ interfaces_before_write = self._get_interfaces_snapshot()
interfaces = self._get_interfaces_section()
interfaces.update(interface_config)
- if not self._write_reticulum_config():
+ if not self._write_reticulum_config(
+ rollback_interfaces=interfaces_before_write
+ ):
return web.json_response(
{
- "message": "Failed to write Reticulum config",
+ "message": (
+ "Failed to write Reticulum config. "
+ "Interface names must not contain '[' or ']' "
+ "(ConfigObj section syntax)."
+ ),
},
status=500,
)
@@ -21296,7 +21347,6 @@ def main():
# Initialize crash recovery system early to catch startup errors
recovery = CrashRecovery()
recovery.install()
- print(_python_jit_status_line())
parser = argparse.ArgumentParser(description="ReticulumMeshChat")
parser.add_argument(
diff --git a/meshchatx/src/backend/community_interfaces.py b/meshchatx/src/backend/community_interfaces.py
index 392668f0..3ffc54f8 100644
--- a/meshchatx/src/backend/community_interfaces.py
+++ b/meshchatx/src/backend/community_interfaces.py
@@ -47,7 +47,12 @@ class CommunityInterfacesManager:
@staticmethod
def _normalize_entry(item: dict[str, Any]) -> dict[str, Any]:
+ from meshchatx.src.backend.interface_editor import InterfaceEditor
+
out = dict(item)
+ name = InterfaceEditor.sanitize_interface_section_name(out.get("name"))
+ if name:
+ out["name"] = name
iface_type = out.get("type")
if iface_type == "BackboneInterface":
remote = (out.get("remote") or out.get("target_host") or "").strip()
diff --git a/meshchatx/src/backend/community_interfaces_directory.py b/meshchatx/src/backend/community_interfaces_directory.py
index 5d2cf831..9bad29e3 100644
--- a/meshchatx/src/backend/community_interfaces_directory.py
+++ b/meshchatx/src/backend/community_interfaces_directory.py
@@ -196,7 +196,9 @@ def transform_directory_rows(rows: list[Any]) -> list[dict[str, Any]]:
if not isinstance(row, dict):
continue
- name = (row.get("name") or "").strip()
+ from meshchatx.src.backend.interface_editor import InterfaceEditor
+
+ name = InterfaceEditor.sanitize_interface_section_name(row.get("name"))
rtype = (row.get("type") or "").lower()
type_name = row.get("typeName") or ""
if not isinstance(type_name, str):
@@ -243,7 +245,7 @@ def transform_directory_rows(rows: list[Any]) -> list[dict[str, Any]]:
if is_backboneish and tid:
out_list.append(
{
- "name": name,
+ "name": name or addr,
"type": "BackboneInterface",
"remote": addr,
"target_port": port_i,
@@ -256,7 +258,7 @@ def transform_directory_rows(rows: list[Any]) -> list[dict[str, Any]]:
if is_backboneish and not tid:
out_list.append(
{
- "name": name,
+ "name": name or addr,
"type": "TCPClientInterface",
"target_host": addr,
"target_port": port_i,
@@ -268,7 +270,7 @@ def transform_directory_rows(rows: list[Any]) -> list[dict[str, Any]]:
if is_tcp_style:
out_list.append(
{
- "name": name,
+ "name": name or addr,
"type": "TCPClientInterface",
"target_host": addr,
"target_port": port_i,
diff --git a/meshchatx/src/backend/data/community_interfaces.json b/meshchatx/src/backend/data/community_interfaces.json
index d40a9333..938e412b 100644
--- a/meshchatx/src/backend/data/community_interfaces.json
+++ b/meshchatx/src/backend/data/community_interfaces.json
@@ -52,7 +52,7 @@
"description": "directory.rns.recipes (user-submitted, online)"
},
{
- "name": "MSK SZAO [HaLow Bridge]",
+ "name": "MSK SZAO (HaLow Bridge)",
"type": "TCPClientInterface",
"target_host": "dreadgurizta.ru",
"target_port": 4242,
diff --git a/meshchatx/src/backend/interface_editor.py b/meshchatx/src/backend/interface_editor.py
index c89b39e2..836f3bc0 100644
--- a/meshchatx/src/backend/interface_editor.py
+++ b/meshchatx/src/backend/interface_editor.py
@@ -106,6 +106,25 @@ class InterfaceEditor:
normalize_rnode_txpower = staticmethod(normalize_rnode_txpower)
validate_rnode_txpower = staticmethod(validate_rnode_txpower)
+ @staticmethod
+ def sanitize_interface_section_name(name: str | None) -> str:
+ """Make a name safe for Reticulum/ConfigObj ``[[section]]`` headers.
+
+ Square brackets break ConfigObj nesting and can leave the in-memory
+ interfaces map dirty after a failed write, blocking later adds.
+ """
+ cleaned = str(name or "").strip()
+ if not cleaned:
+ return ""
+ cleaned = (
+ cleaned.replace("[", "(")
+ .replace("]", ")")
+ .replace("\n", " ")
+ .replace("\r", " ")
+ )
+ cleaned = " ".join(cleaned.split())
+ return cleaned[:128]
+
@staticmethod
def minimum_fixed_mtu() -> int:
mtu = getattr(RNS.Reticulum, "MTU", None)
diff --git a/tests/backend/test_interface_section_names.py b/tests/backend/test_interface_section_names.py
new file mode 100644
index 00000000..a9877424
--- /dev/null
+++ b/tests/backend/test_interface_section_names.py
@@ -0,0 +1,196 @@
+# SPDX-License-Identifier: 0BSD
+
+"""ConfigObj-safe Reticulum interface section names."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import RNS
+from RNS.vendor.configobj import ConfigObj
+
+from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend.community_interfaces import CommunityInterfacesManager
+from meshchatx.src.backend.interface_editor import InterfaceEditor
+
+
+def test_sanitize_interface_section_name_strips_brackets():
+ assert (
+ InterfaceEditor.sanitize_interface_section_name("MSK SZAO [HaLow Bridge]")
+ == "MSK SZAO (HaLow Bridge)"
+ )
+ assert InterfaceEditor.sanitize_interface_section_name(" foo[bar] ") == "foo(bar)"
+ assert InterfaceEditor.sanitize_interface_section_name("") == ""
+ assert InterfaceEditor.sanitize_interface_section_name(None) == ""
+
+
+def test_configobj_write_succeeds_but_reload_fails_for_brackets():
+ """RNS ConfigObj writes bracket section names, then NestingError on reload."""
+ path = Path(tempfile.mktemp(suffix=".cfg"))
+ try:
+ cfg = ConfigObj(str(path))
+ cfg["interfaces"] = {}
+ cfg["interfaces"]["MSK SZAO [HaLow Bridge]"] = {
+ "type": "TCPClientInterface",
+ "target_host": "example.com",
+ "target_port": "4242",
+ }
+ cfg.write()
+ with pytest.raises(Exception):
+ ConfigObj(str(path))
+ finally:
+ path.unlink(missing_ok=True)
+
+
+def test_configobj_accepts_sanitized_section_names():
+ path = Path(tempfile.mktemp(suffix=".cfg"))
+ try:
+ name = InterfaceEditor.sanitize_interface_section_name(
+ "MSK SZAO [HaLow Bridge]"
+ )
+ cfg = ConfigObj(str(path))
+ cfg["interfaces"] = {}
+ cfg["interfaces"][name] = {
+ "type": "TCPClientInterface",
+ "target_host": "example.com",
+ "target_port": "4242",
+ }
+ cfg.write()
+ reloaded = ConfigObj(str(path))
+ assert name in reloaded["interfaces"]
+ finally:
+ path.unlink(missing_ok=True)
+
+
+def test_community_manager_normalizes_bracket_names(tmp_path):
+ doc = {
+ "interfaces": [
+ {
+ "name": "MSK SZAO [HaLow Bridge]",
+ "type": "TCPClientInterface",
+ "target_host": "dreadgurizta.ru",
+ "target_port": 4242,
+ }
+ ]
+ }
+ path = tmp_path / "community_interfaces.json"
+ path.write_text(json.dumps(doc), encoding="utf-8")
+ manager = CommunityInterfacesManager(public_override_path=str(path))
+ assert manager.interfaces[0]["name"] == "MSK SZAO (HaLow Bridge)"
+
+
+class _ConfigDict(dict):
+ def __init__(self, *args, fail_write=False, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.fail_write = fail_write
+ self.write_called = False
+
+ def write(self):
+ self.write_called = True
+ if self.fail_write:
+ raise RuntimeError("NestingError: Cannot compute the section depth")
+ return True
+
+
+async def _find_add_handler(app_instance):
+ for route in app_instance.get_routes():
+ if route.path == "/api/v1/reticulum/interfaces/add" and route.method == "POST":
+ return route.handler
+ return None
+
+
+@pytest.mark.asyncio
+async def test_add_interface_sanitizes_brackets_and_writes(tmp_path):
+ config = _ConfigDict({"reticulum": {}, "interfaces": {}})
+ identity = MagicMock(spec=RNS.Identity)
+ identity.hash = b"test_hash_32_bytes_long_01234567"
+ identity.hexhash = identity.hash.hex()
+ identity.get_private_key.return_value = b"test_private_key"
+
+ with (
+ patch("meshchatx.meshchat.generate_ssl_certificate"),
+ patch("RNS.Reticulum") as mock_rns,
+ patch("RNS.Transport"),
+ patch("LXMF.LXMRouter"),
+ ):
+ mock_reticulum = mock_rns.return_value
+ mock_reticulum.config = config
+ mock_reticulum.configpath = str(tmp_path / "config")
+ mock_reticulum.is_connected_to_shared_instance = False
+ mock_reticulum.transport_enabled.return_value = True
+
+ app = ReticulumMeshChat(
+ identity=identity,
+ storage_dir=str(tmp_path),
+ reticulum_config_dir=str(tmp_path),
+ )
+ handler = await _find_add_handler(app)
+ assert handler is not None
+
+ class Request:
+ @staticmethod
+ async def json():
+ return {
+ "name": "MSK SZAO [HaLow Bridge]",
+ "type": "TCPClientInterface",
+ "target_host": "dreadgurizta.ru",
+ "target_port": 4242,
+ "enabled": True,
+ }
+
+ response = await handler(Request())
+ body = json.loads(response.body)
+ assert response.status == 200, body
+ assert "MSK SZAO (HaLow Bridge)" in config["interfaces"]
+ assert "MSK SZAO [HaLow Bridge]" not in config["interfaces"]
+ assert config.write_called is True
+
+
+@pytest.mark.asyncio
+async def test_failed_write_rolls_back_dirty_interfaces(tmp_path):
+ config = _ConfigDict({"reticulum": {}, "interfaces": {}}, fail_write=True)
+ identity = MagicMock(spec=RNS.Identity)
+ identity.hash = b"test_hash_32_bytes_long_01234567"
+ identity.hexhash = identity.hash.hex()
+ identity.get_private_key.return_value = b"test_private_key"
+
+ with (
+ patch("meshchatx.meshchat.generate_ssl_certificate"),
+ patch("RNS.Reticulum") as mock_rns,
+ patch("RNS.Transport"),
+ patch("LXMF.LXMRouter"),
+ ):
+ mock_reticulum = mock_rns.return_value
+ mock_reticulum.config = config
+ mock_reticulum.configpath = str(tmp_path / "config")
+ mock_reticulum.is_connected_to_shared_instance = False
+ mock_reticulum.transport_enabled.return_value = True
+
+ app = ReticulumMeshChat(
+ identity=identity,
+ storage_dir=str(tmp_path),
+ reticulum_config_dir=str(tmp_path),
+ )
+ handler = await _find_add_handler(app)
+ assert handler is not None
+
+ class Request:
+ @staticmethod
+ async def json():
+ return {
+ "name": "Good Node",
+ "type": "TCPClientInterface",
+ "target_host": "example.com",
+ "target_port": 4242,
+ "enabled": True,
+ }
+
+ response = await handler(Request())
+ body = json.loads(response.body)
+ assert response.status == 500, body
+ assert "Failed to write Reticulum config" in body["message"]
+ assert config["interfaces"] == {}
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────